package cn.xshi.iot.common.mqtt;

import cn.xshi.common.util.ExceptionUtil;
import cn.xshi.common.util.StringUtil;
import cn.xshi.iot.common.vo.MqttMessageDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;
/**
 * @Desc Mqtt发布
 * @Author 邓纯杰
 * @CreateTime 2012-12-12 12:12:12
 */
@Slf4j
@Component
public class PublishMqttClient {

    /*
    消息服务质量，一共有三个：
    0：尽力而为。消息可能会丢，但绝不会重复传输
    1：消息绝不会丢，但可能会重复传输
    2：恰好一次。每条消息肯定会被传输一次且仅传输一次
   */
//    @Value("${spring.mqtt.qos}")
    private int qos =2;

    /**
     * 消息发布
     * @author scj
     **/
    public Boolean publish(MqttMessageDTO mqttMessageDTO) {
        boolean  bool = true;
        MqttClient client = null;
        try {
            if(StringUtil.isEmpty(mqttMessageDTO.getHost())){
                throw new ExceptionUtil("mqtt服务端地址host为空");
            }
            if(StringUtil.isEmpty(mqttMessageDTO.getUserName())){
                throw new ExceptionUtil("mqtt服务端用户名为空");
            }
            if(StringUtil.isEmpty(mqttMessageDTO.getPassword())){
                throw new ExceptionUtil("mqtt服务端密码为空");
            }
            if(StringUtil.isEmpty(mqttMessageDTO.getTopic())){
                throw new ExceptionUtil("发送mqtt topic为空");
            }
            //clientId不能重复所以这里我设置为系统时间
            String clientId = mqttMessageDTO.getClientId();
            // host为主机名，clientId即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientId的保存形式，默认为以内存保存
            client = new MqttClient(mqttMessageDTO.getHost(), clientId, new MemoryPersistence());
            // MQTT的连接设置
            MqttConnectOptions options = new MqttConnectOptions();
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
            options.setCleanSession(true);
            // 设置连接的用户名
            if(StringUtils.isNotEmpty(mqttMessageDTO.getUserName())){
                options.setUserName(mqttMessageDTO.getUserName());
                // 设置连接的密码
                options.setPassword(mqttMessageDTO.getPassword().toCharArray());
            }
            // 连接超时时间
            options.setConnectionTimeout(10);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options.setKeepAliveInterval(20);
            //发布者连接服务
            client.connect(options);
            // 获取TOPIC
            MqttTopic mqttTopic = client.getTopic(mqttMessageDTO.getTopic());
            // 消息内容
            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setQos(qos);
            if(StringUtils.isNotBlank(mqttMessageDTO.getMessage())){
                mqttMessage.setPayload(mqttMessageDTO.getMessage().getBytes());
                //Retained为true时MQ会保留最后一条发送的数据，当断开再次订阅即会接收到这最后一次的数据
                mqttMessage.setRetained(false);
            }else{
                //如果清空上次发送的内容 message为null即可,这样客户端重启订阅 就不能收到你上次发送的结果了
                mqttMessage.setPayload(new byte[0]);
                //Retained为true时MQ会保留最后一条发送的数据，当断开再次订阅即会接收到这最后一次的数据
                mqttMessage.setRetained(true);
            }

            // 订阅消息
            MqttDeliveryToken deliveryToken = mqttTopic.publish(mqttMessage);
            deliveryToken.waitForCompletion();
            bool = deliveryToken.isComplete();
            if (bool) {
//                log.info("发布者发布消息： " + message + " 成功，topic："+topic);
            } else {
                log.error("发布者发布消息： " + mqttMessageDTO.getMessage() + " 失败，topic："+mqttMessageDTO.getTopic());
                deliveryToken.waitForCompletion();
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("发布者发布消息异常： " + mqttMessageDTO.getMessage() + " 失败，topic："+mqttMessageDTO.getTopic()+",exception:"+e);
        }finally {
            try{
                if(null != client){
                    client.disconnect();
                    client.close();
                }
            }catch (Exception e){
                bool = false;
                e.printStackTrace();
                log.error("Finally发布者发布消息异常： " + mqttMessageDTO.getMessage() + " 失败，topic："+mqttMessageDTO.getTopic()+",exception:"+e);
            }
        }
        return bool;
    }

//    /**
//     * 发布
//     * @param topic
//     * @param content
//     * @return
//     */
//    public Boolean publish(String topic, String content) {
////        log.info("MQ===public=== 入参:topic:{};content:{}", topic, content);
//        MqttMessage message = new MqttMessage(content.getBytes());
//        message.setQos(qos);
//        /**
//         * Retained为true时MQ会保留最后一条发送的数据，当断开再次订阅即会接收到这最后一次的数据
//         */
//        message.setRetained(false);
//        try {
//            String clientid= CommonUtil.getClientId();
//            MqttClient mqttClient = this.connect(clientid, username,passWord);
//            // 判定是否需要重新连接
//            if (mqttClient == null || !mqttClient.isConnected() || !mqttClient.getClientId().equals(clientid)) {
//                mqttClient = this.connect(clientid, username,passWord);
//            }
//            mqttClient.publish(topic, message);
////            log.info("Mqtt已发topic: {} - message: {}", topic, message);
//        } catch (Exception e) {
//            e.printStackTrace();
////            log.error("Finally发布者发布消息异常： " + message + " 失败，topic："+topic+",exception:"+e);
//            return false;
//        }
//        return true;
//    }
//
//    /**
//     * 连接
//     * @param clientId
//     * @param userName
//     * @param password
//     * @return
//     * @throws MqttException
//     */
//    public MqttClient connect(String clientId, String userName, String password) throws MqttException {
//        MemoryPersistence persistence = new MemoryPersistence();
//        MqttConnectOptions options = new MqttConnectOptions();
//        options.setCleanSession(true);
//        options.setUserName(userName);
//        options.setPassword(password.toCharArray());
//        options.setConnectionTimeout(10);
//        options.setKeepAliveInterval(20);
//        MqttClient client = new MqttClient(host, clientId, persistence);
//        client.setCallback(new MQTTSubsribeCallback());
//        client.connect(options);
//        return client;
//    }

//    /**
//     *
//     * @param args
//     * @throws ExecutionException
//     * @throws InterruptedException
//     */
//    public static void main(String[] args) throws ExecutionException,InterruptedException {
//        System.out.println("----程序开始运行----");
//        Date date1 = new Date();
//
//        int taskSize = 1;
//        // 创建一个线程池
//        ExecutorService pool = Executors.newFixedThreadPool(taskSize);
//        for (int i = 0; i < taskSize; i++) {
//            PublishMqttClient publishMqttClient = new PublishMqttClient();
//            publishMqttClient.publish("test","测试一下");
//        }
//        // 关闭线程池
//        pool.shutdown();
//        Date date2 = new Date();
//        System.out.println("----程序结束运行----，程序运行时间【"
//                + (date2.getTime() - date1.getTime()) + "毫秒】");
//    }

}
